﻿using CommonModel.Util;
using GrpcPtlSvr;
using KafkaUtil;
using KafkaUtil.Interface;
using NewLife;
using NewLife.Log;
using NewLife.Net;
using NewLife.Threading;
using Newtonsoft.Json;
using PtlSave.Util;
using PtlSvr;
using PtlSvr.Model;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;

namespace Gxrtbtc.Ptl
{
    /// <summary>定义服务端，用于管理所有网络会话</summary>
    public class PtlNetServer : NetServer<PtlNetSession>
    {
        //动态库加载类
        public PtlLoader ptlLoader { get; set; }
        //kafka生产者
        public KafkaProducerOptions producerOptions { get; set; }
        //kafka生产者 错误数据的保存
        public KafkaProducerOptions producerErrOptions { get; set; }
        public ProducerPool _errProducerPool = null;

        //是否使用过滤机制
        public bool filter { get; set; } = false;
        //key为ip:port 标注这些数据发送错误时，需要写到Kafka
        public Dictionary<string, int> filters { get; set; } = new Dictionary<string, int>();

        //对于创建NetSession会话，过滤掉一些，因为有些是无用且会占用很多资源的会话
        public Dictionary<string, int> filterIps { get; set; } = null;

        //连接记录
        public Dictionary<string, TerminalStatus> terminalInfosAll = new Dictionary<string, TerminalStatus>();
        public Dictionary<string, TerminalStatus> terminalInfos = new Dictionary<string, TerminalStatus>();
        //接收包总数
        public ulong RxPackCount = 0;
        public ulong RxPackCountPre = 0;
        //接收包速率 /分钟
        public ulong RxPackRate = 0;
        //出错记录数
        public ulong RxErrPackCount = 0;

        public PtlNetServer(string ignoreIps = "")
        {
            filterIps = new Dictionary<string, int>();
            if (!"".Equals(ignoreIps))
            {
                foreach (var ip in ignoreIps.Split(','))
                {
                    filterIps.Add(ip, 10);
                }
            }

            //kafka生产者池初始化
            _errProducerPool = new ProducerPool
            {
                Name = "ProducerPool",
                producerOptions = producerErrOptions,
                Min = 1,  //最小poolNum个连接
                //Max = maxPoolNum * 2,   //最大maxPoolNum个连接
                Max = 0,   //最大个数。默认100，0表示无上限
                IdleTime = 10,  //空闲时间20秒，超过该空闲时间的连接将被销毁释放，直到留下最后Min个
                AllIdleTime = 60   //完全空闲时间120秒，把最后Min个连接都干掉
            };

            //计算接收包速率 间隔60秒
            new TimerX(GetRxPackTate, "Timer", 1000, 60 * 1000) { Async = true };
        }

        //计算接收包速率
        private void GetRxPackTate(Object state)
        {
            RxPackRate = RxPackCount - RxPackCountPre;
            RxPackCountPre = RxPackCount;
        }

        protected override INetSession OnNewSession(ISocketSession session)
        {
            //过滤掉部分无用的连接 一个IP来源超过10个无用的连接，则判断这个来源无效，可以过滤掉
            if (filterIps.ContainsKey(session.Remote.Address.ToString()))
            {
                if (filterIps[session.Remote.Address.ToString()] >= 10)
                { 
                    session.Dispose();
                    return null;
                }
            }
            else
            {
                filterIps.Add(session.Remote.Address.ToString(), 1);
            }
            return base.OnNewSession(session);
        }
    }

    /// <summary>定义会话。每一个远程连接唯一对应一个网络会话，再次重复收发信息</summary>
    public class PtlNetSession : NetSession<PtlNetServer>
    {
        /***************************基本参数变量********************************************/
        //服务端
        private PtlNetServer ptlNetServer = null;
        //动态库操作，目前默认一个连接只返回一种格式或标准的数据
        //private PtlImpHelper ptlhelper = null;
        private Dictionary<string, PtlImpHelper> ptlhelpers = null;

        //线程池数量
        private int poolNum = 1;
        //最大允许线程池数量
        private int maxPoolNum = 5;
        //最大允许已下发处理数量和已处理数量的差，即阀值
        private int maxGapNum = 100;
        //已下发处理数量和已处理数量的差
        private int numGap = 0;
        //单位时间内发布失败的数量
        private int failNum = 0;
        private int dataCnt = 0;    //单位时间内接收到的数据量

        //kafka生产者池
        private ProducerPool _producerPool = null;

        //是否使用过滤机制
        private bool filter = false;
        private bool dealing = true;

        //数据缓存
        private BlockingCollection<Object[]> dataCache = null; //Redis做消息队列

        private TimerX _timer;
        private Thread dealThead;

        //临时变量，key为长度，作简单判断是否处理
        private Dictionary<int, byte[]> preDatas = null;
        /***************************基本参数变量********************************************/

        /// <summary>客户端连接</summary>
        protected override void OnConnected()
        {
            WriteLog($"{Environment.MachineName} 连接成功!  [{Remote}]\r\n");
            base.OnConnected();

            //初始化
            ptlNetServer = (PtlNetServer)base.Host;
            if (!ptlNetServer.terminalInfos.ContainsKey(Remote.EndPoint.ToString()))
            {
                ptlNetServer.terminalInfos.Add(Remote.EndPoint.ToString(), new TerminalStatus() { Terminalinfo = new TerminalInfo { Ip = Remote.Address.ToString(), Port = uint.Parse(Remote.Port.ToString()), Ptlver = "" }, TerminalData = new TerminalData { RxPackCount = 0, RxPackRate = 0, RxErrPackCount = 0 } });
            }
            if (!ptlNetServer.terminalInfosAll.ContainsKey(Remote.EndPoint.ToString()))
            {
                ptlNetServer.terminalInfosAll.Add(Remote.EndPoint.ToString(), new TerminalStatus() { Terminalinfo = new TerminalInfo { Ip = Remote.Address.ToString(), Port = uint.Parse(Remote.Port.ToString()), Ptlver = "" }, TerminalData = new TerminalData { RxPackCount = 0, RxPackRate = 0, RxErrPackCount = 0 } });
            }

            dataCache = new BlockingCollection<Object[]>();

            ptlhelpers = new Dictionary<string, PtlImpHelper>();
            preDatas = new Dictionary<int, byte[]>();
            filter = ptlNetServer.filter;

            //kafka生产者池初始化
            _producerPool = new ProducerPool
            {
                Name = "ProducerPool",
                producerOptions = ptlNetServer.producerOptions,
                Min = poolNum + 1,  //最小poolNum个连接
                //Max = maxPoolNum * 2,   //最大maxPoolNum个连接
                Max = 0,   //最大个数。默认100，0表示无上限
                IdleTime = 10,  //空闲时间20秒，超过该空闲时间的连接将被销毁释放，直到留下最后Min个
                AllIdleTime = 60   //完全空闲时间120秒，把最后Min个连接都干掉
            };

            //每分钟计算一次
            _timer = new TimerX(SetParam, "Timer", 5 * 1000, 60 * 1000) { Async = true };

            //启动数据处理
            dealThead = new Thread(DealData);
            dealThead.Start();

            WriteLog(" 初始化 poolNum【" + poolNum + "】 最大允许【" + maxPoolNum + "】 阈值【" + maxGapNum + "】 差值【" + numGap + "】");
        }

        /// <summary>客户端断开连接</summary>
        protected override void OnDisconnected()
        {
            dealing = false;
            ptlNetServer.terminalInfos.Remove(Remote.EndPoint.ToString());
            _producerPool.Dispose();
            _timer.Dispose();
            dataCache.Dispose();
            dealThead.TryDispose();

#if DEBUG
            WriteLog("客户端{0}已经断开连接啦", Remote);
#endif

            base.OnDisconnected();
        }

        /// <summary>收到客户端数据</summary>
        /// <param name="e"></param>
        protected override void OnReceive(ReceivedEventArgs e)
        {
            //无用的数据直接丢弃
            if (e.Packet.ToArray().Length < 5)
            {
                //添加记录
                ptlNetServer.filterIps[Remote.Address.ToString()]++;
                return;
            }
            byte[] bt = AddData(e.Packet.ToArray(), Remote);

            string key = TransForm.ByteToHexStringBigEndian(bt.Skip(15).Take(1).ToArray());
            if (!ptlhelpers.ContainsKey(key))
            {
                ptlhelpers.Add(key, ptlNetServer.ptlLoader.matching(bt));
                if (ptlhelpers[key] != null)
                {
                    string ver = ptlhelpers[key].getver();
                    ptlNetServer.terminalInfos[Remote.EndPoint.ToString()].Terminalinfo.Ptlver = ver;
                    ptlNetServer.terminalInfosAll[Remote.EndPoint.ToString()].Terminalinfo.Ptlver = ver;
                    ptlNetServer.filterIps[Remote.Address.ToString()] = 1;
                    WriteLog(Remote + "-找到匹配接口" + ver + " 数据长度" + e.Packet.ToArray().Length + " 数据：" + CommonUtil.byteToHex(e.Packet.ToArray()));
                }
                else
                {
                    WriteLog(Remote + "未找到匹配接口" + " 数据长度" + e.Packet.ToArray().Length + " 数据：" + CommonUtil.byteToHex(e.Packet.ToArray()));
                }
            }

            if (null != ptlhelpers[key])
            {
                ptlNetServer.RxPackCount++;
                ptlNetServer.terminalInfos[Remote.EndPoint.ToString()].TerminalData.RxPackCount++;
                ptlNetServer.terminalInfosAll[Remote.EndPoint.ToString()].TerminalData.RxPackCount++;

                //简单过滤数据，与前一条数据完全一样的不处理
                if (filter)
                {
                    if (!preDatas.ContainsKey(bt.Length))
                    {
                        preDatas.Add(bt.Length, bt);

                        //数据写入缓存
                        lock (dataCache)
                        {
                            dataCache.Add(new Object[] { key, bt });
                        }
                    }
                    //与前一条数据不一样才写入kafka，否则丢弃
                    else if (!Enumerable.SequenceEqual(bt, preDatas[bt.Length]))
                    {
                        preDatas[bt.Length] = bt;

                        //数据写入缓存
                        lock (dataCache)
                        {
                            dataCache.Add(new Object[] { key, bt });
                        }
                    }
                }
                //不过滤
                else
                { 
                    //数据写入缓存
                    lock (dataCache)
                    {
                        dataCache.Add(new Object[] { key, bt });
                    }
                }
            }
        }

        /// <summary>出错</summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        protected override void OnError(Object sender, ExceptionEventArgs e)
        {
#if DEBUG
            WriteLog(Remote + "[{0}]错误：{1}", e.Action, e.Exception?.GetTrue().Message);
#endif

            base.OnError(sender, e);
        }

        /*
         * 计算并初始化kafka及设置线程池参数
         */
        private void SetParam(Object state)
        {
            if (numGap > maxGapNum && poolNum < maxPoolNum)
            {
                poolNum++;
                ThreadPool.SetMaxThreads(poolNum, poolNum);

                WriteLog(" 当前 poolNum【" + poolNum + "】 最大允许【" + maxPoolNum + "】 阈值【" + maxGapNum + "】 差值【" + numGap + "】");
            }

            if (failNum > 0)
            {
                WriteError(Remote + " 发布信息到Kafka失败！failNum=" + failNum);
                failNum = 0;
            }

            if (dataCnt > 0)
            {
                XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " " + Remote.Address.ToString() + ":" + Remote.Port.ToString() + " volume " + dataCnt + "/MIN");
                dataCnt = 0;
            }
            ptlNetServer.terminalInfos[Remote.EndPoint.ToString()].TerminalData.RxPackRate = ulong.Parse(dataCnt.ToString());
            ptlNetServer.terminalInfosAll[Remote.EndPoint.ToString()].TerminalData.RxPackRate = ulong.Parse(dataCnt.ToString());
        }

        /*
         * 缓存数据处理线程
         */
        private void DealData()
        {
            ThreadPool.SetMaxThreads(poolNum, poolNum);
            while (dealing)
            {
                try
                {
                    while (dataCache.Count > 0)
                    {
                        Object[] data = null;
                        lock (dataCache)
                        {
                            data = dataCache.Take();
                        }
                        if (data != null)
                        { 
                            ThreadPool.QueueUserWorkItem(ConvertPool, data);
                            dataCnt++;
                            numGap++;
                        }
                        Thread.Sleep(0);
                    }
                }
                catch { }
                finally
                {
                    Thread.Sleep(0);
                }
            }
        }

        /*
         * 缓存数据处理，解析并写入kafka
         */
        private void ConvertPool(Object data)
        {
            IClientProducer producer = _producerPool.Get();
            Object[] dataTmp = (Object[])data;
            try
            {
                var sw = Stopwatch.StartNew();
                string jsondata = ptlhelpers[dataTmp[0].ToString()].convert((byte[])dataTmp[1]);
                int n = new Random().Next(0, poolNum);
                if (jsondata != null && !"".Equals(jsondata))
                {
                    producer.Publish(jsondata, result =>
                    {
                        if (result != null)
                        {
                            numGap--;
                        }
                        else
                        {
                            failNum++;
                        }
                    });//发送到kafka中

                    //producer.PublishAsync(jsondata, result =>
                    //{
                    //    if (result != null)
                    //    {
                    //        numGap--;
                    //    }
                    //    else
                    //    {
                    //        failNum++;
                    //    }
                    //});//发送到kafka中 异步
                }
                else
                {
                    if (ptlNetServer.filters.ContainsKey(Remote.EndPoint.ToString()))
                    {
                        SetErr((byte[])dataTmp[1]);
                    }

                    ptlNetServer.RxErrPackCount++;
                    ptlNetServer.terminalInfos[Remote.EndPoint.ToString()].TerminalData.RxErrPackCount++;
                    ptlNetServer.terminalInfosAll[Remote.EndPoint.ToString()].TerminalData.RxErrPackCount++;
                }
            }
            catch (Exception e)
            {
                //异常时
                WriteLog("用时：{0}", "处理异常，" + e.Message + " data:" + CommonUtil.byteToHex((byte[])dataTmp[1]));
            }
            finally
            {
                _producerPool.Put(producer);
            }
        }

        /*
         * 错误数据处理
         */
        private void SetErr(byte[] data)
        {
            string jsondata = "";
            IClientProducer errProducer = ptlNetServer._errProducerPool.Get();
            try
            {
                Dictionary<string, object> resultDict = new Dictionary<string, object>();//发送下去的总json数据
                Dictionary<string, string> ptlDict = new Dictionary<string, string>();
                ptlDict.Add("ptlver", ptlNetServer.terminalInfosAll[Remote.EndPoint.ToString()].Terminalinfo.Ptlver);
                ptlDict.Add("ts", CommonUtil.GetFormatDatetime());//时间戳
                ptlDict.Add("ptlnode", "");//协议转换节点

                Dictionary<string, string> addDict = new Dictionary<string, string>();
                addDict.Add("ip", Remote.Address.ToString());//ip地址
                addDict.Add("port", Remote.Port.ToString());//端口

                resultDict.Add("ptlinfo", ptlDict);
                resultDict.Add("addr", addDict);
                resultDict.Add("datas", CommonUtil.byteToHex(data));
                jsondata = JsonConvert.SerializeObject(resultDict);
            }
            catch
            {
                jsondata = CommonUtil.byteToHex(data);
            }
            finally
            {
                errProducer.Publish(jsondata, result =>
                {
                    if (result != null)
                    {
                        numGap--;
                    }
                    else
                    {
                        failNum++;
                    }
                });//发送到kafka中
                ptlNetServer._errProducerPool.Put(errProducer);
            }
        }

        /*
         * 数据包前新增IP、端口、时间戳等信息
         */
        private byte[] AddData(byte[] data, NetUri Remote)
        {
            //IP占4位，端口占2位，时间戳占8位, 协议类别1位 共15位
            byte[] newData = new byte[15 + data.Length];

            //IP放最前面4位
            byte[] tmpHost = CommonUtil.FormatIpToBytes(Remote.Address.ToString());
            Buffer.BlockCopy(tmpHost, 0, newData, 0, 4);

            //端口2位放IP后
            byte[] tmpPort = CommonUtil.IntToByteArray(Remote.Port, false, false);
            Buffer.BlockCopy(tmpPort, 0, newData, 4, 2);

            //时间戳8位
            byte[] tmpTimeStamp = CommonUtil.LongToBytes(long.Parse(CommonUtil.GetTimeStamp()), false);
            Buffer.BlockCopy(tmpTimeStamp, 0, newData, 6, 8);

            //把协议类别放在时间戳后面
            Buffer.BlockCopy(new byte[1], 0, newData, 14, 1);

            //原始数据在最后
            Buffer.BlockCopy(data, 0, newData, 15, data.Length);
            return newData;
        }
    }
}